package com.swak.reactivex.transport;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.swak.Constants;
import com.swak.exception.RemotingException;

/**
 * 心跳处理
 * 
 * @author lifeng
 */
public class HeartbeatHandler extends ChannelHandlerDelegate {

	private static final Logger logger = LoggerFactory.getLogger(HeartbeatHandler.class);

	public HeartbeatHandler(ChannelHandler handler) {
		super(handler);
	}

	@Override
	public void sent(Channel channel, Object message) throws RemotingException {
		if (isHeartbeatRequest(message)) {
			if (logger.isDebugEnabled()) {
				logger.debug("Sent heartbeat request in thread " + Thread.currentThread().getName());
			}
			return;
		}
		if (isHeartbeatResponse(message)) {
			if (logger.isDebugEnabled()) {
				logger.debug("Sent heartbeat response in thread " + Thread.currentThread().getName());
			}
			return;
		}
		super.sent(channel, message);
	}

	@Override
	public void received(Channel channel, Object message) throws RemotingException {
		if (isHeartbeatRequest(message)) {
			Message req = (Message) message;
			if (req.isTwoWay()) {
				channel.send(Message.forResponse(req).addOption(Message.Heartbeat));
				if (logger.isInfoEnabled()) {
					int heartbeat = channel.url().getParameter(Constants.Heartbeat, 0);
					if (logger.isDebugEnabled()) {
						logger.debug("Received heartbeat from remote channel " + channel.remoteAddress()
								+ ", cause: The channel has no data-transmission exceeds a heartbeat period"
								+ (heartbeat > 0 ? ": " + heartbeat + "ms" : ""));
					}
				}
			}
			return;
		}
		if (isHeartbeatResponse(message)) {
			if (logger.isDebugEnabled()) {
				logger.debug("Receive heartbeat response in thread " + Thread.currentThread().getName());
			}
			return;
		}
		super.received(channel, message);
	}

	private boolean isHeartbeatRequest(Object message) {
		return message instanceof Message && ((Message) message).isRequest() && ((Message) message).isHeartbeat();
	}

	private boolean isHeartbeatResponse(Object message) {
		return message instanceof Message && ((Message) message).isResponse() && ((Message) message).isHeartbeat();
	}
}